package com.apps.sdses.flink141.demo;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * @author jiwei
 * @description
 * @date 2023/10/26 18:10
 */
public class Datagen2MySQL_zbill_biz_run {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //设置时区
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));

        //Source
        tableEnv.executeSql("CREATE TABLE source_zbill_source_run (\n" +
                "     id BIGINT,\n" +
                "     periodNo STRING,\n" +
                "     source STRING,\n" +
                "     channel STRING ,\n" +
                "     hisTradeNo STRING ,\n" +
                "     platTradeNo STRING,\n" +
                "     sourceTradeNo STRING,\n" +
                "     platRefundNo STRING,\n" +
                "     bizRefundNo STRING,\n" +
                "     sourceRefundNo STRING,\n" +
                "     tradeType STRING,\n" +
                "     amount DECIMAL(10, 2),\n" +
                "     completeTime TIMESTAMP(3)," +
                "     PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (\n" +
                "     'connector' = 'datagen',\n" +
//                "     'rows-per-second' = '1',\n" +
                "     'number-of-rows' = '5',\n" +
                "     'fields.id.min'='1',\n" +
                "     'fields.id.max'='10',\n" +
                "     'fields.periodNo.length'='5',\n" +
                "     'fields.source.length'='5',\n" +
                "     'fields.channel.length'='5',\n" +
                "     'fields.hisTradeNo.length'='5',\n" +
                "     'fields.platTradeNo.length'='5',\n" +
                "     'fields.sourceTradeNo.length'='5',\n" +
                "     'fields.platRefundNo.length'='5',\n" +
                "     'fields.bizRefundNo.length'='5',\n" +
                "     'fields.sourceRefundNo.length'='5',\n" +
                "     'fields.platRefundNo.length'='5',\n" +
                "     'fields.tradeType.length'='3')");

        //Sink
        tableEnv.executeSql("CREATE TABLE sink_zbill_source_run (\n" +
                "     id BIGINT,\n" +
                "     periodNo STRING,\n" +
                "     source STRING,\n" +
                "     channel STRING ,\n" +
                "     hisTradeNo STRING ,\n" +
                "     platTradeNo STRING,\n" +
                "     sourceTradeNo STRING,\n" +
                "     platRefundNo STRING,\n" +
                "     bizRefundNo STRING,\n" +
                "     sourceRefundNo STRING,\n" +
                "     tradeType STRING,\n" +
                "     amount DECIMAL(10, 2),\n" +
                "     completeTime TIMESTAMP(3)," +
                "     PRIMARY KEY(id) NOT ENFORCED\n" +
                "     ) WITH (\n" +
                "     'connector' = 'jdbc',\n" +
                "     'url' = 'jdbc:mysql://192.168.102.155:33066/db_datacube_test',\n" +
                "     'driver' = 'com.mysql.cj.jdbc.Driver',\n" +
                "     'table-name' = 'zbill_biz_run',\n" +
                "     'username' = 'datacube_test',\n" +
                "     'password' = 'Gbd#XERA@19d1ZtD')");

        //Insert
        tableEnv.executeSql("insert into sink_zbill_source_run\n" +
                "SELECT * FROM source_zbill_source_run");
    }
}
